/**
 * Copyright (C) 2009+ Spark Zheng
 *
 * @file        src/pipeline.c 
 * @date        2009-03-25
 * @brief	general_datastream
 * @see         dserver
 * @history
 */

#include <dlib/lock.h>
#include <dlib/str.h>
#include <dlib/mempool.h>

#include "dserver_log.h"
#include "dserver_errno.h"
#include "filter.h"
#include "pipeline.h"

static inline int _pipeline_link_filter(pipeline_node_t *pipeline, filter_node_t *filter);
static inline int _pipeline_unlink_filter(pipeline_node_t *pipeline);

static inline int _pipeline_link_filter(pipeline_node_t *pipeline, filter_node_t *filter)
{
	filter_node_t *tpos = NULL;
	list_node_t *pos = NULL;

	/* if (pipeline->p_filter_nr == 0) {
		pipeline->p_filter_nr = 1;
		LIST_HEAD_INIT(&pipeline->p_filter_head);
	} else {
		pipeline->p_filter_nr += 1;
	} */

	pipeline->p_filter_nr += 1;

	LIST_FOREACH_ENTRY(tpos, pos, &pipeline->p_filter_head, p_node) {
		if (tpos->f_settings.f_filter_id > filter->f_settings.f_filter_id) {
			break;
		} else if (tpos->f_settings.f_filter_id ==
				filter->f_settings.f_filter_id) {
			DLOG_ERR(LOG_MAIN, "same filter id, parameters error.");
			return -E_ARGS_ILLEGAL;
		}
	}
	if (pos == &pipeline->p_filter_head) {
		LIST_ADD_TAIL(&pipeline->p_filter_head, &filter->p_node);
	} else {
		LIST_INSERT_BEFORE(&tpos->p_node, &filter->p_node);
	}

	/*
	LIST_FOREACH_ENTRY(tpos, pos, &pipeline->p_filter_head, p_node) {
		DLOG_DEBUG(LOG_MAIN, "filter[%s:%d:%d].", tpos->f_name,
				tpos->f_settings.f_pipeline_id, tpos->f_settings.f_filter_id);
	}
	*/

	/* link and run filter */
	DLOG_DEBUG(LOG_MAIN, "dserver run filter[%s] for tpool[%d:%d].",
			filter->f_name, filter->f_settings.f_pmin, filter->f_settings.f_pmax);
	filter->f_tpool = dlib_tpool_init(filter->f_settings.f_pmin,
			filter->f_settings.f_pmax, filter->f_ops.f_stream_init,
			(void *)filter, filter->f_ops.f_stream_exit);
	if (filter->f_tpool == NULL) {
		DLOG_ERR(LOG_MAIN, "tpool_init error for filter %s.", filter->f_name);
		return -E_PROC_THREAD;
	}

	dlib_tpool_settask(filter->f_tpool, filter->f_ops.f_stream_worker, (void *)filter);
	dlib_tpool_settotask(filter->f_tpool, filter->f_settings.f_timeout,
			filter->f_ops.f_stream_timeout, (void *)filter);
	dlib_tpool_setdetask(filter->f_tpool, filter->f_ops.f_stream_finish, (void *)filter);

	if (dlib_tpool_timeout_async(filter->f_tpool) < 0) {
		DLOG_ERR(LOG_MAIN, "tpool_timeout_async error for filter %s.",
				filter->f_name);
		return -E_PROC_THREAD;
	}
	if (dlib_tpool_retrieve_async(filter->f_tpool) < 0) {
		DLOG_ERR(LOG_MAIN, "tpool_retrieve_async error for filter %s.",
				filter->f_name);
		return -E_PROC_THREAD;
	}

	DLOG_INFO(LOG_MAIN, "dserver link filter[%s] success.", filter->f_name);

	return RET_SUCCESS;
}

static inline int _pipeline_unlink_filter(pipeline_node_t *pipeline)
{
	filter_node_t *tpos = NULL;
	list_node_t *pos = NULL;

	if (pipeline->p_filter_nr == 0) {
		return RET_SUCCESS;
	}

	LIST_FOREACH_ENTRY(tpos, pos, &pipeline->p_filter_head, p_node) {
		dlib_tpool_wait(tpos->f_tpool);
		dlib_tpool_exit(tpos->f_tpool);
	}

	return RET_SUCCESS;
}

int pipeline_init(dserver_t *dc)
{
	int ret = RET_SUCCESS;
	uint16_t i;
	uint16_t max_id = 0;
	filter_node_t *tpos = NULL;
	list_node_t *pos = NULL;
	dserver_filter_t *filters;
	pipeline_node_t *pipeline;
	dserver_pipeline_t *pipelines;
	
	DLOG_DEBUG(LOG_MAIN, "dserver enters in pipeline_init function.");

	if (dc == NULL || dc->filters == NULL) {
		DLOG_ERR(LOG_MAIN, "pipeline_init parameters error.");
		return -E_ARGS_ILLEGAL;
	}

	pipelines = dmalloc(sizeof(*pipelines));
	dc->pipelines = pipelines;
	filters = dc->filters;
	
	//dlib_lock_init(&pipelines->p_lock);
	pipelines->p_nr = 0;

	/* get max_id for O(1) to access pipeline */
	LIST_FOREACH_ENTRY(tpos, pos, &filters->f_head, f_node) {
		if (tpos->f_settings.f_pipeline_id > max_id) {
			max_id = tpos->f_settings.f_pipeline_id;
		}
	}
	max_id += 1;
	if (max_id > MAX_PIPELINE_ID) {
		DLOG_ERR(LOG_MAIN, "pipeline_init filter parameters error.");
		return -E_ARGS_ILLEGAL;
	}
	pipelines->p_max_id = max_id;

	pipelines->p_nodes = dmalloc(max_id * sizeof(pipeline_node_t));

	/* init pipeline nodes */
	for (i = 0; i < max_id; i++) {
		pipeline = pipelines->p_nodes + i;
		pipeline->p_filter_nr = 0;
		LIST_HEAD_INIT(&pipeline->p_filter_head);
	}

	/* link filter into pipeline */
	LIST_FOREACH_ENTRY(tpos, pos, &filters->f_head, f_node) {
		pipeline = pipelines->p_nodes + (tpos->f_settings.f_pipeline_id);
		ret = _pipeline_link_filter(pipeline, tpos);
		if (ret < 0) {
			DLOG_ERR(LOG_MAIN, "pipeline_init filter link error[%d].", ret);
			return ret;
		}
	}/* foreach filter */

	return ret;
}

void pipeline_exit(dserver_t *dc)
{
	uint16_t i;
	pipeline_node_t *pipeline;
	dserver_pipeline_t *pipelines;

	DLOG_DEBUG(LOG_MAIN, "dserver enters in pipeline_exit function.");

	if (dc == NULL || dc->pipelines == NULL || dc->filters == NULL) {
		return;
	}
	
	pipelines = dc->pipelines;
	if (pipelines->p_nodes != NULL) {
		for (i = 0; i < pipelines->p_max_id; i++) {
			pipeline = pipelines->p_nodes + i;
			(void)_pipeline_unlink_filter(pipeline);
		}
		dfree(pipelines->p_nodes);
		pipelines->p_nodes = NULL;
	}
	//dlib_lock_exit(&pipelines->p_lock);

	return;
}

